package tableApi;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

public class tabletest05_StreamAndTable {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String filePath = "src/main/resources/sensor.txt";
        tableEnv.connect(new FileSystem().path(filePath))
                .withFormat(new Csv())//表的文件格式
                .withSchema(new Schema()//表的数据结构,属性名是可以变更的，但是顺序时不可以变换的
                        .field("id", DataTypes.STRING())
                        .field("timeStamp", DataTypes.BIGINT())
                        .field("temperature", DataTypes.DOUBLE()))
                .createTemporaryTable("sensor");

        Table inputTable = tableEnv.from("sensor");

        //表--->流
        DataStream<Row> resultStream = tableEnv.toAppendStream(inputTable, Row.class);
        DataStream<Tuple2<Boolean, Row>> RetrackStream = tableEnv.toRetractStream(inputTable, Row.class);

        //流--->表
        Table table = tableEnv.fromDataStream(resultStream);
        Table table1 = tableEnv.fromDataStream(RetrackStream, "id timestamp as ts,temperature");

        //基于DataStream创建视图
        tableEnv.createTemporaryView("ts", resultStream);
        tableEnv.createTemporaryView("ts2", resultStream, "id ,temperature,timestamp as ts");

        //基于Table创建视图
        tableEnv.createTemporaryView("ts3", inputTable);
    }
}
